package studio.raptor.databus.client;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Properties;
import javax.jms.Connection;
import javax.jms.MessageConsumer;
import javax.jms.Topic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import studio.raptor.cmdb.Config;
import studio.raptor.cmdb.ConfigService;
import studio.raptor.databus.ActiveMQConnectionFactory;
import studio.raptor.databus.ActiveMQSession;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Session;
import studio.raptor.databus.client.config.BrokerGroup;

/**
 * 消费者配置，包含消费者所属业务中心标志，消费的主题及消费者个性化配置。<br><br>
 * <p>
 * <p>
 * 消费者默认配置如下：<br>
 * 会话不开启事务；<br>
 * 消费者自动ACK消息模式；<br>
 * 开启批量ACK功能；<br>
 * 批量ACK超时时间30s；<br>
 * 消息重发次数不受限制；<br>
 * 每个消费者预取消息数量200；<br>
 * 两次重连之间最长的时间间隔为10s(从10毫秒开始指数增加到10s)<br>
 * failover后的尝试重连次数<br>
 * </P>
 */
public class DatabusConsumerBuilder {

  private static Logger log = LoggerFactory.getLogger(DatabusConsumerBuilder.class);

  /**
   * 会话是否为事务性，消费者默认为否
   */
  private boolean transactedSession = false;

  /**
   * 消息ACK模式，默认值(1)为自动确认
   */
  private int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
  /**
   * 开启批量ACK优化
   */
  private boolean optimizeAcknowledge = false;
  /**
   * ACK消息超时时间，默认30s--当不开启ACK优化时，此值无意义
   */
  private int optimizeAcknowledgeTimeOut = 30000;
  /**
   * 消息重发次数，默认5，表示直到消息被ACK，否则一直重发。0表示不重传,-1表示一直重发
   */
  private int maximumRedeliveries = 5;
  /**
   * 默认预取值设定为200
   */
  private int prefetchSize = 1;
  /**
   * 两次重连之间最长时间间隔，10s.
   */
  private int maxReconnectDelay = 10000;
  /**
   * failover机制中的最大重连次数。-1表示一直重连。
   */
  private int maxReconnectAttempts = -1;

  /**
   * 客户端启动时最大重连次数，默认为5
   */
  private int startupMaxReconnectAttempts = 5;

  /**
   * 能否支持订阅主题模式，默认为false。一般只有databus下的应用才会开启
   */
  private boolean subscribeMode = false;
  /**
   * 通用其他配置项
   */
  private String otherURLConfig = "&";


  private List<BrokerGroup> brokerGroups = new ArrayList<BrokerGroup>();


  private DatabusConfig databusConfig;

  public DatabusConsumerBuilder(DatabusConfig databusConfig) {
    this.databusConfig = databusConfig;
    this.brokerGroups = databusConfig.getBrokerGroups();
  }

  /**
   * 创建死信队列消费者-默认情况下为手动ACK消息
   *
   * @param eventName 事件名称
   * @param eventKindId 事件种类ID
   */
  public DatabusConsumer createDLQ(String eventName, Integer eventKindId) throws JMSException {
    String consumerSelector = null;
    return createDLQ(eventName, eventKindId,consumerSelector);
  }
  /**
   * 创建订阅死信主题队列消费者-默认情况下为手动ACK消息
   *
   * @param eventName 事件名称
   * @param eventKindId 事件种类ID
   * @param clientId 持久化订阅者客户端ID
   * @param durableSubscriberName 持久化订阅者名称
   */
  public DatabusConsumer createDLQ(String eventName, Integer eventKindId,
      String clientId,String durableSubscriberName) throws JMSException {
    String consumerSelector = null;
    return createDLQ(eventName,eventKindId,consumerSelector,clientId,durableSubscriberName);
  }

  /**
   * 创建死信队列消费者-默认情况下为手动ACK消息,携带消息选择器
   * <p>
   * 方法的参数consumerSelector为当前消费者的消息过滤条件，其遵循SQL92语法定义。如：version = 12 and env = 'dev',
   * 表示当前消费者会从服务端过滤消息属性version为12并且env为dev的消息，不满足此过滤条件的消息消费者不会从服务端获取。
   * <br/>
   * 选择器区分消息属性类型，若设置消息属性时，version设置为字符串属性，则以上例子中的过滤条件将无法选择到此消息
   * </p>
   *
   * @param eventName 事件名称
   * @param eventKindId 事件种类ID
   * @param consumerSelector 消费者消息选择器，参数遵循SQL92语法
   */
  public DatabusConsumer createDLQ(String eventName, Integer eventKindId, String consumerSelector)
      throws JMSException {
    String clientId = null;
    String durableSubscriberName = null;
    boolean isDLQ = true;
    boolean isClientACK = true;
    return doCreate(eventName,eventKindId,isDLQ,isClientACK,consumerSelector,clientId,durableSubscriberName,null);
  }

  /**
   * 创建订阅死信主题队列消费者-默认情况下为手动ACK消息,携带消息选择器
   * <p>
   * 方法的参数consumerSelector为当前消费者的消息过滤条件，其遵循SQL92语法定义。如：version = 12 and env = 'dev',
   * 表示当前消费者会从服务端过滤消息属性version为12并且env为dev的消息，不满足此过滤条件的消息消费者不会从服务端获取。
   * <br/>
   * 选择器区分消息属性类型，若设置消息属性时，version设置为字符串属性，则以上例子中的过滤条件将无法选择到此消息
   * </p>
   *
   * @param eventName 事件名称
   * @param eventKindId 事件种类ID
   * @param consumerSelector 消费者消息选择器，参数遵循SQL92语法
   * @param clientId 持久化订阅者客户端ID
   * @param durableSubscriberName 持久化订阅者名称
   */
  public DatabusConsumer createDLQ(String eventName, Integer eventKindId, String consumerSelector,
      String clientId,String durableSubscriberName)
      throws JMSException {
    boolean isDLQ = true;
    boolean isClientACK = true ;
    return doCreate(eventName, eventKindId, isDLQ, isClientACK, consumerSelector,clientId,durableSubscriberName,null);
  }

  /**
   * 创建需要客户端手工确认的消费者，此消费者应该使用{@link DatabusConsumer#receiveUnACKMessage(long)} 方法获取消息
   *
   * @param eventName 事件名称
   * @param eventKindId 事件种类ID
   */
  public DatabusConsumer createClientACK(String eventName, Integer eventKindId)
      throws JMSException {
    String clientId = null,durableSubscriberName = null,consumerSelector = null;
    boolean isDLQ = false,isClientACK = true;
    return doCreate(eventName, eventKindId, isDLQ, isClientACK, consumerSelector, clientId,
        durableSubscriberName,null);
  }

  /**
   * 创建需要客户端手工确认的持久化订阅者，此消费者应该使用{@link DatabusConsumer#receiveUnACKMessage(long)} 方法获取消息
   *
   * @param eventName 事件名称
   * @param eventKindId 事件种类ID
   * @param clientId 持久订阅者客户端ID
   * @param durableSubscriberName 持久订阅者名称
   */
  public DatabusConsumer createClientACK(String eventName, Integer eventKindId, String clientId,
      String durableSubscriberName) throws JMSException {
    String consumerSelector = null;
    boolean isDLQ = false,isClientACK = true;
    return doCreate(eventName, eventKindId, isDLQ, isClientACK, consumerSelector, clientId,
        durableSubscriberName,null);
  }

  /**
   * 创建需要客户端手工确认的消费者，此消费者应该使用{@link DatabusConsumer#receiveUnACKMessage(long)} 方法获取消息
   * <p>
   * 方法的参数consumerSelector为当前消费者的消息过滤条件，其遵循SQL92语法定义。如：version = 12 and env = 'dev',
   * 表示当前消费者会从服务端过滤消息属性version为12并且env为dev的消息，不满足此过滤条件的消息消费者不会从服务端获取。
   * <br/>
   * 选择器区分消息属性类型，若设置消息属性时，version设置为字符串属性，则以上例子中的过滤条件将无法选择到此消息
   * </p>
   *
   * @param eventName 事件名称
   * @param eventKindId 事件种类ID
   * @param consumerSelector 消息过滤器
   */
  public DatabusConsumer createClientACK(String eventName, Integer eventKindId,
      String consumerSelector) throws JMSException {
    boolean isDLQ = false,isClientACK = true;
    String clientId = null, durableSubscriberName = null;
    return doCreate(eventName, eventKindId, isDLQ, isClientACK, consumerSelector, clientId,
        durableSubscriberName,null);
  }

  /**
   * 创建需要客户端手工确认的持久化订阅者，此消费者应该使用{@link DatabusConsumer#receiveUnACKMessage(long)} 方法获取消息
   * <p>
   * 方法的参数consumerSelector为当前消费者的消息过滤条件，其遵循SQL92语法定义。如：version = 12 and env = 'dev',
   * 表示当前消费者会从服务端过滤消息属性version为12并且env为dev的消息，不满足此过滤条件的消息消费者不会从服务端获取。
   * <br/>
   * 选择器区分消息属性类型，若设置消息属性时，version设置为字符串属性，则以上例子中的过滤条件将无法选择到此消息
   * </p>
   *
   * @param eventName 事件名称
   * @param eventKindId 事件种类ID
   * @param consumerSelector 消息过滤器
   * @param clientId 持久订阅者客户端ID
   * @param durableSubscriberName 持久订阅者名称
   */
  public DatabusConsumer createClientACK(String eventName, Integer eventKindId,
      String consumerSelector, String clientId, String durableSubscriberName) throws JMSException {
    boolean isDLQ = false;
    boolean isClientACK = true;

    return doCreate(eventName, eventKindId, isDLQ, isClientACK, consumerSelector, clientId,
        durableSubscriberName,null);
  }

  /**
   * 创建默认配置的消费者-自动ACK消息
   *
   * @param eventName 事件名称
   * @param eventKindId 事件种类ID
   * @return 返回消费者
   */
  public DatabusConsumer create(String eventName, Integer eventKindId) throws JMSException {
    String clientId = null;
    String durableSubscriberName = null;
    return create(eventName, eventKindId, clientId, durableSubscriberName);
  }

  /**
   * 创建默认配置的消费者-自动ACK消息(仅限databus内部使用)
   *
   * @param eventName 事件名称
   * @param eventKindId 事件种类ID
   * @param clientId 持久化订阅者客户ID
   * @param durableSubscriberName 持久化订阅者名称
   * @return 返回消费者
   */
  public DatabusConsumer create(String eventName, Integer eventKindId, String clientId,
      String durableSubscriberName) throws JMSException {
    boolean isDLQ = false;
    boolean isClientACK = false;
    String consumerSelector = null;
    return doCreate(eventName, eventKindId, isDLQ, isClientACK, consumerSelector, clientId,
        durableSubscriberName,null);
  }

  /**
   * 创建默认配置的消费者-自动ACK消息
   * <p>
   * 方法的参数consumerSelector为当前消费者的消息过滤条件，其遵循SQL92语法定义。如：version = 12 and env = 'dev',
   * 表示当前消费者会从服务端过滤消息属性version为12并且env为dev的消息，不满足此过滤条件的消息消费者不会从服务端获取。
   * <br/>
   * 选择器区分消息属性类型，若设置消息属性时，version设置为字符串属性，则以上例子中的过滤条件将无法选择到此消息
   * </p>
   *
   * @param eventName 事件名称
   * @param eventKindId 事件种类ID
   * @param consumerSelector 消息事件选择器
   * @return 返回消费者
   */
  public DatabusConsumer create(String eventName, Integer eventKindId, String consumerSelector)
      throws JMSException {
    String clientId = null;
    String durableSubscriberName = null;
    return create(eventName, eventKindId, consumerSelector, clientId, durableSubscriberName);
  }


  /**
   * 创建默认配置的消费者-自动ACK消息(仅限databus内部使用)
   * <p>
   * 方法的参数consumerSelector为当前消费者的消息过滤条件，其遵循SQL92语法定义。如：version = 12 and env = 'dev',
   * 表示当前消费者会从服务端过滤消息属性version为12并且env为dev的消息，不满足此过滤条件的消息消费者不会从服务端获取。
   * <br/>
   * 选择器区分消息属性类型，若设置消息属性时，version设置为字符串属性，则以上例子中的过滤条件将无法选择到此消息
   * </p>
   *
   * @param eventName 事件名称
   * @param eventKindId 事件种类ID
   * @param consumerSelector 消息事件选择器
   * @param clientId 持久化订阅者客户ID
   * @param durableSubscriberName 持久化订阅者名称
   * @return 返回消费者
   */
  public DatabusConsumer create(String eventName, Integer eventKindId, String consumerSelector,
      String clientId, String durableSubscriberName) throws JMSException {
    boolean isDLQ = false;
    boolean isClientACK = false;
    return doCreate(eventName, eventKindId, isDLQ, isClientACK, consumerSelector, clientId,
        durableSubscriberName,null);
  }



  /**
   * 真实创建消费者，当且仅当clientId和durableSubscriberName都有值且非空字符串时创建持久订阅者，否则创建普通消费者
   * @param eventName 消费者消费的事件名称
   * @param eventKindId 消费者消费的事件种类ID
   * @param isDLQ 是否是消费死信队列
   * @param isClientACK 消费者是否是自动ACK
   * @param consumerSelector 消息过滤器，若为null则表示不创建携带消息的过滤器
   * @param clientId 连接的客户端ID 若为null则创建普通队列消费者
   * @param durableSubscriberName 持久化订阅者名称，若为null则创建普通队列消费者
   * @param groupName 应用内部组名称，若为空或者则表示不需要创建
   * @return
   * @throws JMSException
   */
  private DatabusConsumer doCreate(String eventName, Integer eventKindId,
      boolean isDLQ, boolean isClientACK, String consumerSelector,
      String clientId, String durableSubscriberName,String groupName) throws JMSException {

    DatabusConsumer databusConsumer = new DatabusConsumer();
    databusConsumer.setDatabusConfig(databusConfig);
    checkEventInfo(eventName, eventKindId, databusConsumer);

    try {
      initConfig(isClientACK);
    } catch (IOException e) {
      log.error(e.getMessage());
      throw new JMSException(e.getMessage());
    }
    ActiveMQConnectionFactory connectionFactory = buildConnectionFactory(this.databusConfig,
        eventName, eventKindId);
    Connection connection = connectionFactory.createConnection();
    databusConsumer.setConnection(connection);
    boolean shouldCreateDurableSubscriber = shouldCreateDurableSubscriber(clientId,
        durableSubscriberName);
    if (shouldCreateDurableSubscriber) {
      if( isDLQ ){
        clientId = clientId+DatabusConfig.TDLQ_FIX;
      }
      connection.setClientID(clientId);
    }
    databusConsumer.getConnection().start();
    Session session = connection
        .createSession(this.isTransactedSession(), this.getAcknowledgeMode());
    databusConsumer.setSession(session);
    databusConsumer.setEventSeq(eventKindId);
    databusConsumer.setEventName(eventName);

    MessageConsumer consumer = null;
    if (shouldCreateDurableSubscriber) {
      consumer = createDurableSubscriber(session, eventName, eventKindId, durableSubscriberName,
          isDLQ, consumerSelector);
    } else {
      consumer = createConsumer(session, eventName, eventKindId, isDLQ, consumerSelector,groupName);
    }
    databusConsumer.setMessageConsumer(consumer);
    return databusConsumer;
  }

  /**
   * 依据加载配置方式初始化消费者配置
   */
  private void initConfig(boolean isClientACK) throws IOException {
    if (isClientACK) {
      this.setAcknowledgeMode(ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE);
    } else {
      this.setAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
    }
    if (databusConfig.getLoadType() == DatabusConfigLoadType.LOAD_TYPE_CMDB) {
      //从配置中心加载消费者配置
      Config config = ConfigService
          .getConfig(DatabusConfig.CLIENT_CONFIG_PRE + databusConfig.getBizKey());
      this.setTransactedSession(
          config.getBooleanProperty("consumer.transactedSession", this.isTransactedSession()));
      this.setOptimizeAcknowledge(
          config.getBooleanProperty("consumer.optimizeAcknowledge", this.isOptimizeAcknowledge()));
      this.setOptimizeAcknowledgeTimeOut(config
          .getIntProperty("consumer.optimizeAcknowledgeTimeOut",
              this.getOptimizeAcknowledgeTimeOut()));
      this.setMaximumRedeliveries(
          config.getIntProperty("consumer.maximumRedeliveries", this.getMaximumRedeliveries()));
      this.setPrefetchSize(config.getIntProperty("consumer.prefetchSize", this.getPrefetchSize()));
      this.setMaxReconnectDelay(
          config.getIntProperty("consumer.maxReconnectDelay", this.getMaxReconnectDelay()));
      this.setMaxReconnectAttempts(
          config.getIntProperty("consumer.maxReconnectAttempts", this.getMaxReconnectAttempts()));
      this.setStartupMaxReconnectAttempts(config
          .getIntProperty("consumer.startupMaxReconnectAttempts",
              this.getStartupMaxReconnectAttempts()));
      this.setSubscribeMode(
          config.getBooleanProperty("consumer.subscribeMode", this.isSubscribeMode())
      );
      this.setOtherURLConfig(
          config.getProperty("consumer.otherURLConfig", this.getOtherURLConfig()));
    } else {
      InputStream in = Thread.currentThread().getContextClassLoader()
          .getResourceAsStream(this.databusConfig.CONFIG_LOCAL_DIR + "consumer.properties");
      Properties properties = new Properties();
      properties.load(in);

      this.setTransactedSession(
          "true".equals(properties.getProperty("consumer.transactedSession")));
      this.setOptimizeAcknowledge(
          "true".equals(properties.getProperty("consumer.optimizeAcknowledge")));
      this.setOptimizeAcknowledgeTimeOut(
          Integer.valueOf(properties.getProperty("consumer.optimizeAcknowledgeTimeOut")));
      this.setMaximumRedeliveries(
          Integer.valueOf(properties.getProperty("consumer.maximumRedeliveries")));
      this.setPrefetchSize(Integer.valueOf(properties.getProperty("consumer.prefetchSize")));
      this.setMaxReconnectDelay(
          Integer.valueOf(properties.getProperty("consumer.maxReconnectDelay")));
      this.setMaxReconnectAttempts(
          Integer.valueOf(properties.getProperty("consumer.maxReconnectAttempts")));
      this.setStartupMaxReconnectAttempts(
          Integer.valueOf(properties.getProperty("consumer.startupMaxReconnectAttempts")));
      this.setSubscribeMode(
          "true".equals(properties.getProperty("consumer.subscribeMode"))
      );
      this.setOtherURLConfig(properties.getProperty("consumer.otherURLConfig"));
    }

  }

  boolean isTransactedSession() {
    return transactedSession;
  }

  /**
   * 设定会话是否为事务性，消费者默认为否
   */
  void setTransactedSession(boolean transactedSession) {
    this.transactedSession = transactedSession;
  }

  int getAcknowledgeMode() {
    return acknowledgeMode;
  }

  boolean isSubscribeMode() {
    return subscribeMode;
  }

  void setSubscribeMode(boolean subscribeMode) {
    this.subscribeMode = subscribeMode;
  }

  /**
   * 消息ACK模式
   *
   * @param acknowledgeMode 默认(1)为自动确认模式。2为手工批量ACK，4为手工单条消息确认。
   */
  void setAcknowledgeMode(int acknowledgeMode) {
    this.acknowledgeMode = acknowledgeMode;
  }

  boolean isOptimizeAcknowledge() {
    return optimizeAcknowledge;
  }

  /**
   * 设置是否开启批量ACK操作(当且仅当session不是事务性且ACK模式为自动确认时，开启此功能才有意义)
   *
   * @param optimizeAcknowledge 默认开启
   */
  void setOptimizeAcknowledge(boolean optimizeAcknowledge) {
    this.optimizeAcknowledge = optimizeAcknowledge;
  }

  int getOptimizeAcknowledgeTimeOut() {
    return optimizeAcknowledgeTimeOut;
  }

  /**
   * 设置批量ACK超时时间(当且仅当session不是事务性且ACK模式为自动确认时，此值才有意义)
   *
   * @param optimizeAcknowledgeTimeOut ACK超时时间，单位毫秒
   */
  void setOptimizeAcknowledgeTimeOut(int optimizeAcknowledgeTimeOut) {
    this.optimizeAcknowledgeTimeOut = optimizeAcknowledgeTimeOut;
  }

  int getMaximumRedeliveries() {
    return maximumRedeliveries;
  }

  /**
   * 设置消息最大重发次数(建议使用默认值)
   *
   * @param maximumRedeliveries 默认为5
   */
  void setMaximumRedeliveries(int maximumRedeliveries) {
    this.maximumRedeliveries = maximumRedeliveries;
  }

  int getPrefetchSize() {
    return prefetchSize;
  }

  /**
   * 设置消费者的预取消息数量
   *
   * @param prefetchSize
   */
  void setPrefetchSize(int prefetchSize) {
    this.prefetchSize = prefetchSize;
  }

  int getMaxReconnectDelay() {
    return maxReconnectDelay;
  }

  /**
   * 设置最大重来延迟毫秒数
   *
   * @param maxReconnectDelay 默认10s
   */
  void setMaxReconnectDelay(int maxReconnectDelay) {
    this.maxReconnectDelay = maxReconnectDelay;
  }

  int getMaxReconnectAttempts() {
    return maxReconnectAttempts;
  }

  /**
   * 设置failover机制最大重连次数
   *
   * @param maxReconnectAttempts 默认值为-1，表示一直尝试重连。0表示不重连
   */
  void setMaxReconnectAttempts(int maxReconnectAttempts) {
    this.maxReconnectAttempts = maxReconnectAttempts;
  }

  String getOtherURLConfig() {
    return otherURLConfig;
  }

  /**
   * 设置其他参数，默认情况下不使用此项
   */
  void setOtherURLConfig(String otherURLConfig) {
    this.otherURLConfig = otherURLConfig;
  }

  int getStartupMaxReconnectAttempts() {
    return startupMaxReconnectAttempts;
  }

  /**
   * 设置客户端启动时尝试连接服务器的最大次数，默认为5
   *
   * @param startupMaxReconnectAttempts 客户端启动时尝试连接服务器的最大次数
   */
  void setStartupMaxReconnectAttempts(int startupMaxReconnectAttempts) {
    this.startupMaxReconnectAttempts = startupMaxReconnectAttempts;
  }

  /**
   * 将消费者客户端配置转换为URL参数
   */
  public String transform2URLConfig() {
    StringBuilder urlConfig = new StringBuilder();
    urlConfig.append("jms.optimizeAcknowledge=" + this.isOptimizeAcknowledge());
    urlConfig.append("&jms.optimizeAcknowledgeTimeOut=" + this.getOptimizeAcknowledgeTimeOut());
    urlConfig.append("&jms.redeliveryPolicy.maximumRedeliveries=" + this.getMaximumRedeliveries());
    urlConfig.append("&maxReconnectDelay=" + this.getMaxReconnectDelay());
    urlConfig.append("&maxReconnectAttempts=" + this.getMaxReconnectAttempts());
    urlConfig.append("&startupMaxReconnectAttempts=" + this.getStartupMaxReconnectAttempts());
    urlConfig.append(this.getOtherURLConfig());
    return urlConfig.toString();
  }

  /**
   * 将消费者的消息目的地配置转换
   */
  public String transform2DestinationConfig() {
    StringBuilder destinationURLConfig = new StringBuilder();
    destinationURLConfig.append("consumer.prefetchSize=" + this.getPrefetchSize());
    return destinationURLConfig.toString();
  }


  /**
   * 通过判断clientId和durableSubscriberName判断是否是需要创建持久化订阅者
   */
  private boolean shouldCreateDurableSubscriber(String clientId, String durableSubscriberName) {
    boolean result = false;
    if (!Objects.equals(null, clientId) &&
        !Objects.equals("", clientId) &&
        !Objects.equals(null, durableSubscriberName) &&
        !Objects.equals("", durableSubscriberName)) {
      result = true;
    }
    return result;
  }

  /**
   * 创建持久化订阅者，订阅者的消息目标指向虚拟主题
   *
   * @param consumerSelector 消息选择器
   */
  private MessageConsumer createDurableSubscriber(Session session, String eventName,
      Integer eventKindId, String durableSubscriberName, boolean isDLQ, String consumerSelector)
      throws JMSException {
    if ( !this.isSubscribeMode() ){
      throw new JMSException("当前应用未授权订阅消息模式!请在CMDB配置中心确认是否被授权！");
    }
    MessageConsumer consumer = null;

    StringBuilder topicNameSbd = new StringBuilder();
    topicNameSbd.append(DatabusConfig.VIRTUAL_TOPIC_PRE);
    topicNameSbd.append(eventName);
    topicNameSbd.append("_");
    topicNameSbd.append(TopicSalt.createSalt(eventName, eventKindId));
    topicNameSbd.append("_");
    topicNameSbd.append(eventKindId);
    String topicName = topicNameSbd.toString();
    if (isDLQ) {
      //创建消费死信队列的持久订阅者，主题名称和订阅者名称需要区分
      topicName = DatabusConfig.TDLQ_PRE + topicName;
      durableSubscriberName = durableSubscriberName +DatabusConfig.TDLQ_FIX;
    }
    Topic topic = session.createTopic(topicName + "?" + this.transform2DestinationConfig());
    if (!Objects.equals(null, consumerSelector) && !Objects.equals("", consumerSelector)) {
      consumer = session
          .createDurableSubscriber(topic, durableSubscriberName, consumerSelector, false);
    } else {
      consumer = session.createDurableSubscriber(topic, durableSubscriberName);
    }
    return consumer;
  }


  /**
   * 创建消费指定队列的消费者
   * @param session
   * @param eventName 事件名称
   * @param eventKindId 事件kidnId
   * @param isDLQ 是否死信队列
   * @param consumerSelector 消息选择器
   * @param groupName 应用内消息组名，若为NULL或者空串表示不需要创建
   * @return MessageConsumer
   * @throws JMSException
   */
  private MessageConsumer createConsumer(Session session, String eventName,
      Integer eventKindId, boolean isDLQ, String consumerSelector,String groupName) throws JMSException {
    String groupQueueName = "";
    MessageConsumer consumer = null;
    if (isDLQ) {
      //若为死信队列
      groupQueueName = DatabusConfig.DLQ_PRE;
    }
    groupName = (Objects.equals(null,groupName)||Objects.equals("",groupName))?".":"&"+groupName+".";
    groupQueueName +=
        DatabusConfig.GROUP_QUEUE_NAME_PRE +
            databusConfig.getBizKey() + groupName+
            DatabusConfig.VIRTUAL_TOPIC_PRE + eventName + "_" + TopicSalt
            .createSalt(eventName, eventKindId) + "_" + eventKindId;
    Destination destination = session
        .createQueue(groupQueueName + "?" + this.transform2DestinationConfig());
    if (!Objects.equals(null, consumerSelector) && !Objects.equals("", consumerSelector)) {
      consumer = session.createConsumer(destination, consumerSelector);
    } else {
      consumer = session.createConsumer(destination);
    }
    return consumer;
  }

  /**
   * 创建组消费者--从死信队列消费
   * @param eventName
   * @param eventKindId
   * @param groupName
   * @return
   * @throws JMSException
   */
  public DatabusConsumer createDLQGroupConsumer(String eventName,Integer eventKindId,String groupName)
      throws JMSException {
    String selector = null ;
    return createDLQGroupConsumer(eventName,eventKindId,groupName,selector);

  }
  /**
   * 创建组消费者-消费死信队列携带消息过滤器
   * @param eventName
   * @param eventKindId
   * @param groupName
   * @param selector
   * @return
   * @throws JMSException
   */
  public DatabusConsumer createDLQGroupConsumer(String eventName,
      Integer eventKindId,String groupName,String selector)
      throws JMSException {
    // check groupName should not be null and current bizKey have permission to use groupName
    if(!checkInnerGroupConsumer(groupName)){
      throw new JMSException("您无权限从组["+groupName+"]消费消息");
    }
    boolean isSingleConsumer = this.databusConfig.getInnerGroupInfo().get(groupName).booleanValue();
    boolean isDLQ = true;
    boolean isClientACK = true;
    if(isSingleConsumer){
      //当前组下只能有一个消费者，即走持久化订阅者的方式
      String clientId = groupName ;
      String durableName = groupName ;
      return doCreate(eventName,eventKindId,isDLQ,isClientACK,selector,clientId,durableName,groupName);
    }else{
      //组下允许有多个消费者，走普通订阅queue的方式，clientId和durable名称为空
      return doCreate(eventName,eventKindId,isDLQ,isClientACK,selector,null,null,groupName);
    }

  }

  /**
   * 创建组消费者-手动ACK消息
   * @param eventName
   * @param eventKindId
   * @param groupName
   * @return
   * @throws JMSException
   */
  public DatabusConsumer createGroupConsumerClientACK(String eventName,Integer eventKindId,String groupName)
      throws JMSException {
    String selector = null ;
    return createGroupConsumerClientACK(eventName,eventKindId,groupName,selector);

  }
  /**
   * 创建组消费者，手动ACK消息-携带消息过滤器
   * @param eventName
   * @param eventKindId
   * @param groupName
   * @param selector
   * @return
   * @throws JMSException
   */
  public DatabusConsumer createGroupConsumerClientACK(String eventName,
      Integer eventKindId,String groupName,String selector)
      throws JMSException {
    // check groupName should not be null and current bizKey have permission to use groupName
    if(!checkInnerGroupConsumer(groupName)){
      throw new JMSException("您无权限从组["+groupName+"]消费消息");
    }
    boolean isSingleConsumer = this.databusConfig.getInnerGroupInfo().get(groupName).booleanValue();
    boolean isDLQ = false;
    boolean isClientACK = true;

    if(isSingleConsumer){
      String clientId = groupName ;
      String durableName = groupName ;
      //当前组下只能有一个消费者，即走持久化订阅者的方式clientId和durable名称为空均设置为groupName
      return doCreate(eventName,eventKindId,isDLQ,isClientACK,selector,clientId,durableName,groupName);
    }else{
      //组下允许有多个消费者，走普通订阅queue的方式，clientId和durable名称为空
      return doCreate(eventName,eventKindId,isDLQ,isClientACK,selector,null,null,groupName);
    }

  }

  /**
   * 创建组消费者
   * @param eventName
   * @param eventKindId
   * @param groupName
   * @return
   * @throws JMSException
   */
  public DatabusConsumer createGroupConsumer(String eventName,Integer eventKindId,String groupName)
      throws JMSException {
    return createGroupConsumer(eventName,eventKindId,groupName,null);

  }
  /**
   * 创建组消费者-携带消息过滤器
   * @param eventName
   * @param eventKindId
   * @param groupName
   * @param selector
   * @return
   * @throws JMSException
   */
  public DatabusConsumer createGroupConsumer(String eventName,
      Integer eventKindId,String groupName,String selector)
      throws JMSException {
    // check groupName should not be null and current bizKey have permission to use groupName
    if(!checkInnerGroupConsumer(groupName)){
      throw new JMSException("您无权限从组["+groupName+"]消费消息");
    }
    boolean isSingleConsumer = this.databusConfig.getInnerGroupInfo().get(groupName).booleanValue();
    boolean isDLQ = false;
    boolean isClientACK = false;
    if(isSingleConsumer){
      String clientId = groupName ;
      String durableName = groupName ;
      //当前组下只能有一个消费者，即走持久化订阅者的方式
      return doCreate(eventName,eventKindId,isDLQ,isClientACK,selector,clientId,durableName,groupName);
    }else{
      //组下允许有多个消费者，走普通订阅queue的方式
      return doCreate(eventName,eventKindId,isDLQ,isClientACK,selector,null,null,groupName);

    }

  }

  /**
   * 校验事件名称、kindId及当前消费者是否有权订阅此事件
   */
  private void checkEventInfo(String eventName, Integer eventKindId,
      DatabusConsumer databusConsumer)
      throws JMSException {
    if (Objects.equals(null, eventName) || Objects.equals("", eventName)) {
      throw new JMSException("您所需消费的事件名称为空，请正确填写!");
    }
    if (Objects.equals(null, eventKindId)) {
      throw new JMSException("您所需消费的事件kindId为空，请正确填写!");
    }
    String topicSalt = TopicSalt.createSalt(eventName, eventKindId);
    if (!databusConsumer.checkRecieveAuthorization(eventKindId, eventName)) {
      throw new JMSException("您无权订阅事件:[" + eventName + "-" + eventKindId + "],若需订阅请先申请");
    }
  }

  /**
   * 获取并配置连接工厂类
   *
   * @param databusConfig 配置
   * @param eventName 事件名称
   * @param eventKindId 事件kindId
   */
  private ActiveMQConnectionFactory buildConnectionFactory(DatabusConfig databusConfig,
      String eventName, Integer eventKindId)
      throws JMSException {
    ActiveMQConnectionFactory connectionFactory = null;
    Integer brokerGroupId =
        this.databusConfig.getBizPubSubEventInfo().getSub().get(eventName + "_" + eventKindId);
    String userName = "";
    String password = "";
    String brokerURL = DatabusConfig.LOAD_CONFIG_FAILURE;
    for (BrokerGroup brokerGroup : this.brokerGroups) {
      if (brokerGroupId.intValue() == brokerGroup.getBrokerGroupId()) {
        userName = brokerGroup.getUserName();
        password = brokerGroup.getPassword();
        brokerURL = brokerGroup.getBrokerURL();
        break;
      }
    }
    if (DatabusConfig.LOAD_CONFIG_FAILURE.equals(brokerURL)) {
      throw new JMSException("当前事件消息未配置发送目的地或配置了错误的目的地");
    }
    // 配置连接工厂类
    connectionFactory = new ActiveMQConnectionFactory();
    connectionFactory.setUserName(userName);
    connectionFactory.setPassword(password);
    String customizedBrokerURL = this.transform2URLConfig();
    connectionFactory.setBrokerURL(brokerURL + "?" + customizedBrokerURL);
    return connectionFactory;
  }

  /**
   * 校验组名称是否为空或者当前应用是否有权限使用组名称
   * @param groupName 组名称
   * @return
   */
  private boolean checkInnerGroupConsumer(String groupName){
    //从配置中心拿到
    if( Objects.equals(null,groupName) || Objects.equals("",groupName) ){
      throw new NullPointerException(" groupName should not be null ");
    }
    if( groupName.contains(".") ){
      throw new IllegalArgumentException(" groupName should not contain [.] ");
    }
    return this.databusConfig.getInnerGroupInfo().containsKey(groupName);

  }


}
